package com.apps.sdses.flink141.source;

import com.fasterxml.jackson.core.type.TypeReference;
import com.fasterxml.jackson.databind.ObjectMapper;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.source.RichSourceFunction;

import java.io.FileInputStream;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.Map;

/**
 * author: codejiwei
 * date: 2023/6/21
 * desc:
 **/
public class ConfigSource extends RichSourceFunction<Map<String, Object>> {
    private boolean running = true;
    private final String configPath;
    private final String FIELDS = "fields";
    private ArrayList array;

    public ConfigSource(String configPath) {
        this.configPath = configPath;
    }


    @Override
    public void open(Configuration parameters) throws Exception {
        FileInputStream fileInputStream = new FileInputStream(configPath);
        ObjectMapper mapper = new ObjectMapper();
        Map<String, Object> map = mapper.readValue(fileInputStream, new TypeReference<Map<String, Object>>() {
        });
        array = (ArrayList)map.get(FIELDS);
    }

    @Override
    public void run(SourceContext<Map<String, Object>> sourceContext) throws Exception {


        while (running) {
            Map<String, Object> data = new HashMap<>();
            for (Object field : array) {

            }

        }
    }

    @Override
    public void cancel() {
        running = false;
    }
}
